import org.apache.spark.sql.SparkSession


/**
  * @author dinghao 
  * @create 2021-08-19 13:01 
  * @message
  */
object JDBCSource {
  def main(args: Array[String]): Unit = {
//    mysqlSource()
//    hiveSource()
//    kingbaseSource()
//    dmSource()
    //------OK-------
    igniteSource()
//    druidSource()
  }

  def mysqlSource():Unit ={
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("MysqlSource")
      //.config("spark.some.config.option", "some-value")
      .getOrCreate()

    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("mysql.jdbc.driver"))
      .option("url", ConnectMessage.get("mysql.jdbc.url"))
      .option("user", ConnectMessage.get("mysql.jdbc.user"))
      .option("password", ConnectMessage.get("mysql.jdbc.passwd"))
      .option("dbtable", ConnectMessage.get("mysql.jdbc.sql"))
//      .option("query","select * from xmanager.xm_service")
      .load.show
    //    jdbcData.select("hostname","ip").filter("hostname != 'node1'").show
  }

  def hiveSource():Unit={
    val spark = SparkSession.builder.appName("HiveSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("hive.jdbc.driver"))
      .option("url", ConnectMessage.get("hive.jdbc.url"))
      .option("user", ConnectMessage.get("hive.jdbc.user"))
//      .option("password", ConnectMessage.get("hive.jdbc.passwd"))
      .option("query",ConnectMessage.get("hive.jdbc.sql"))
      .load.show
  }

  def kingbaseSource():Unit={
    val spark = SparkSession.builder.appName("HiveSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("kb.jdbc.driver"))
      .option("url", ConnectMessage.get("kb.jdbc.url"))
      .option("user", ConnectMessage.get("kb.jdbc.user"))
      .option("password", ConnectMessage.get("kb.jdbc.passwd"))
      .option("query",ConnectMessage.get("kb.jdbc.sql"))
      .load.show
  }

  def dmSource():Unit={
    val spark = SparkSession.builder.appName("DMSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("dm.jdbc.driver"))
      .option("url", ConnectMessage.get("dm.jdbc.url"))
      .option("user", ConnectMessage.get("dm.jdbc.user"))
      .option("password", ConnectMessage.get("dm.jdbc.passwd"))
      .option("query",ConnectMessage.get("dm.jdbc.sql"))
      .load.show
  }

  def KADBSource():Unit={
    val spark = SparkSession.builder.appName("KADBSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("kadb.jdbc.driver"))
      .option("url", ConnectMessage.get("kadb.jdbc.url"))
      .option("user", ConnectMessage.get("kadb.jdbc.user"))
      .option("password", ConnectMessage.get("kadb.jdbc.passwd"))
      .option("query", ConnectMessage.get("kadb.jdbc.sql"))
      .load.show
  }

  def igniteSource():Unit={
    val spark = SparkSession.builder.appName("IgniteSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("driver",ConnectMessage.get("ignite.jdbc.driver"))
      .option("url", ConnectMessage.get("ignite.jdbc.url"))
      .option("query", ConnectMessage.get("ignite.jdbc.sql"))
      .load.show
  }

  def druidSource():Unit={
    val spark = SparkSession.builder.appName("IgniteSource").master("local[*]").getOrCreate()
    spark.read
      .format("jdbc")
      .option("url",ConnectMessage.get("druid.url"))
      .option("query",ConnectMessage.get("druid.query"))
      .load.show()
  }

}
/*
属性名称	默认值	含义
url		要连接的JDBC URL，可以 在 URL 中 指定 特定于源的连接属性。
dbtable		应该读取或写入的 JDBC 表
query		将数据读入 Spark 的查询语句
driver	“com.mysql.jdbc.Driver”	用于连接到此 URL 的 JDBC 驱动程序的类名
numPartitions		表读取和写入中可用于并行的最大分区数，这同时确定了最大并发的JDBC连接数。
partitionColumn, lowerBound, upperBound		如果指定了任一选项，则必须指定全部选项。此外，还必须指定numPartitions。partitionColumn必须是表中的数字，日期或时间戳列。注意：lowerBound 和 upperBound 仅用于决定分区步幅，而不是用于过滤表中的行。因此，表中的所有行都将被分区并返回。 这些选项仅适用于读操作。
queryTimeout	0	超时时间（单位：秒），零 意味着没有限制。
fetchsize		用于确定每次往返要获取的行数（例如，Oracle是10行），这 可以 用于 提升 JDBC 驱动程序的性能。此选项仅适用于读。
batchsize	1000	JDBC批处理大小，用于确定每次往返要插入的行数。 这 可以 用于 提升 JDBC 驱动程序的性能。此选项仅适用于写。
isolationLevel	READ_UNCOMMITTED	事务隔离级别，适用于当前连接。它可以是 NONE，READ_COMMITTED，READ_UNCOMMITTED，REPEATABLE_READ 或 SERIALIZABLE 之一，对应于 JDBC的Connection 对象定义的标准事务隔离级别，默认值为 READ_UNCOMMITTED。此选项仅适用于写。
sessionInitStatement		在向远程数据库打开每个数据库会话之后，在开始读取数据之前，此选项将执行自定义SQL语句（或PL / SQL块）。 使用它来实现会话初始化，例如：option("sessionInitStatement", """BEGIN execute immediate 'alter session set "_serial_direct_read"=true'; END;""")
truncate	false	当启用SaveMode.Overwrite时，此选项会导致 Spark 截断现有表，而不是删除并重新创建它。这样更高效，并且防止删除表元数据（例如，索引）。但是，在某些情况下，例如新数据具有不同的 schema 时，它将无法工作。此选项仅适用于写。
cascadeTruncate	false	如果JDBC数据库（目前为 PostgreSQL和Oracle）启用并支持，则此选项允许执行TRUNCATE TABLE t CASCADE（在PostgreSQL的情况下，仅执行TRUNCATE TABLE t CASCADE以防止无意中截断表）。这将影响其他表，因此应谨慎使用。此选项仅适用于写。
createTableOptions		此选项允许在创建表时设置特定于数据库的表和分区选项（例如，CREATE TABLE t (name string) ENGINE=InnoDB）。此选项仅适用于写。
createTableColumnTypes		创建表时要使用的数据库列数据类型而不是默认值。（例如：name CHAR（64），comments VARCHAR（1024））。指定的类型应该是有效的 spark sql 数据类型。 此选项仅适用于写。
customSchema		用于从JDBC连接器读取数据的自定义 schema。例如，id DECIMAL(38, 0), name STRING。您还可以指定部分字段，其他字段使用默认类型映射。 例如，id DECIMAL（38,0）。列名应与JDBC表的相应列名相同。用户可以指定Spark SQL的相应数据类型，而不是使用默认值。 此选项仅适用于读。
pushDownPredicate	true	用于 启用或禁用 谓词下推 到 JDBC数据源的选项。默认值为 true，在这种情况下，Spark会尽可能地将过滤器下推到JDBC数据源。否则，如果设置为 false，则不会将过滤器下推到JDBC数据源，此时所有过滤器都将由Spark处理。
 */